package it.tukano.collections;

import it.tukano.log.Log;
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.EOFException;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.logging.Level;
import java.util.logging.Logger;

public class PersistentMapLogger<K extends Serializable, T extends Serializable> {
    private final int
            PUT = 0,
            REMOVE = 1,
            CLEAR = 2,
            PUT_ALL = 3;

    private final LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>();
    private DataOutputStream out;
    private final File file;
    private boolean loggerOnline;
    private Thread loggerThread;

    /**
     * Instance initializer
     */
    public PersistentMapLogger(File file) {
        this.file = file;
    }

    public Map<K, T> readLog() {
        final Map<K, T> map = new LinkedHashMap<K, T>();
        if(!file.exists()) return map;
        
        InputStream in = null;
        try {
            in = new FileInputStream(file);
            DataInputStream din = new DataInputStream(new BufferedInputStream(in));
            readLog(din, map);
        } catch (EOFException ex) {
            Log.message("end of log");
        } catch (Exception ex) {
            Logger.getLogger(PersistentMapLogger.class.getName()).log(Level.SEVERE, null, ex);
        } finally {
            if (in != null) {
                try {
                    in.close();
                } catch (IOException ex) {
                    Logger.getLogger(PersistentMapLogger.class.getName()).log(Level.SEVERE, null, ex);
                }
            }
        }
        return map;
    }

    private void readLog(DataInputStream in, Map<K, T> map) throws Exception {
        for(int code = in.read(); code != -1; code = in.read()) {
            readLog(code, in, map);
        }
    }

    private void readLog(int code, DataInputStream in, Map<K, T> map) throws Exception {
        Log.message("Reading entry:", code);
        switch(code) {
            case CLEAR: readClear(in, map); break;
            case PUT: readPut(in, map); break;
            case REMOVE: readRemove(in, map); break;
            case PUT_ALL: readPutAll(in, map); break;
        }
    }

    private void readClear(DataInputStream in, Map<K, T> map) throws Exception {
        map.clear();
    }

    @SuppressWarnings("unchecked")
    private void readPut(DataInputStream in, Map<K, T> map) throws Exception {
        K key = readKey(in);
        if(key != null) {
            T value = (T) readSerializedObject(in);
            map.put(key, value);
        } else {
            Log.message("null object");
        }
    }

    private void readRemove(DataInputStream in, Map<K, T> map) throws Exception {
        K key = readKey(in);
        map.remove(key);
    }

    @SuppressWarnings("unchecked")
    private void readPutAll(DataInputStream in, Map<K, T> map) throws Exception {
        Serializable datamap = readSerializedObject(in);
        if(datamap == null) {
            Log.message("Null object.");
        } else {
            map.putAll((Map<K, T>) datamap);
        }
    }

    @SuppressWarnings("unchecked")
    private K readKey(DataInputStream in) throws Exception {
        return (K) readSerializedObject(in);
    }

    private Serializable readSerializedObject(DataInputStream in) throws Exception {
        byte[] data = readByteArray(in);
        return data.length == 0 ? null : (Serializable) new ObjectInputStream(new ByteArrayInputStream(data)).readObject();
    }

    private DataOutputStream getDataOutputStream() {
        if(out == null) {
            out = createOutputStream();
        }
        return out;
    }

    public void put(final K key, final T value) {
        startLogger();

        queue.offer(new Runnable() {

            public void run() {
                DataOutputStream dout = getDataOutputStream();
                try {
                    dout.write(PUT);
                    writeKey(key, dout);
                    writeSerializable(value, dout);
                } catch(Exception ex) {
                    throw new RuntimeException(ex);
                }
            }
        });
    }

    private void writeSerializable(Serializable value, DataOutputStream out) throws IOException {
        ByteArrayOutputStream buffer = new ByteArrayOutputStream();
        ObjectOutputStream obj = new ObjectOutputStream(buffer);
        obj.writeObject(value);
        obj.flush();
        obj.close();
        writeBytes(buffer.toByteArray(), out);
    }

    private void writeKey(K key, DataOutputStream out) throws IOException {
        writeSerializable(key, out);
    }

    private void writeBytes(byte[] data, DataOutputStream out) throws IOException {
        out.writeInt(data.length);
        out.write(data);
        out.flush();
    }

    @SuppressWarnings("unchecked")
    public void remove(final Object key) {
        startLogger();
        
        queue.offer(new Runnable() {

            public void run() {
                try {
                    DataOutputStream out = getDataOutputStream();
                    out.write(REMOVE);
                    writeKey((K) key, out);
                } catch(Exception ex) {
                    throw new RuntimeException(ex);
                }
            }
        });
    }

    public void putAll(final Map<? extends K, ? extends T> m) {
        startLogger();
        
        queue.offer(new Runnable() {

            public void run() {
                try {
                    DataOutputStream out = getDataOutputStream();
                    out.write(PUT_ALL);
                    writeSerializable((Serializable) m, out);
                } catch(Exception ex) {
                    throw new RuntimeException(ex);
                }
            }
        });
    }

    public void clear() {
        startLogger();

        queue.offer(new Runnable() {

            public void run() {
                try {
                    getDataOutputStream().write(CLEAR);
                } catch (IOException ex) {
                    throw new RuntimeException(ex);
                }
            }
        });
    }

    private byte[] readByteArray(DataInputStream in) throws IOException {
        int len = in.readInt();
        Log.message("Reading ", len, " bytes");
        byte[] data = new byte[len];
        in.readFully(data);
        return data;
    }

    private DataOutputStream createOutputStream() {
        try {
            FileOutputStream fout = new FileOutputStream(file, true);
            BufferedOutputStream buff = new BufferedOutputStream(fout);
            return new DataOutputStream(buff);
        } catch(IOException ex) {
            throw new RuntimeException(ex);
        }
    }

    public void dispose() {
        if(out != null) {
            try {
                out.close();
            } catch (IOException ex) {
                Logger.getLogger(PersistentMapLogger.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
    }

    private void startLogger() {
        if(loggerOnline) return;
        loggerOnline = true;
        loggerThread = new Thread("PersistentMapLogger Thread") {

            @Override
            public void run() {
                while(true) {
                    try {
                        Runnable task = queue.take();
                        if(task != null) {
                            task.run();
                        }
                        try {
                            getDataOutputStream().flush();
                        } catch (IOException ex) {
                            Logger.getLogger(PersistentMapLogger.class.getName()).log(Level.SEVERE, null, ex);
                        }
                    } catch (InterruptedException ex) {
                        Logger.getLogger(PersistentMapLogger.class.getName()).log(Level.INFO, "logger dies");
                        break;
                    }
                }
                Logger.getLogger(PersistentMapLogger.class.getName()).log(Level.INFO, "logger thread offline");
                PersistentMapLogger.this.dispose();
            }
        };
        loggerThread.start();
    }

    public void waitForEnd() {
        queue.offer(new Runnable() {

            public void run() {
                if(loggerThread != null) {
                    loggerThread.interrupt();
                }
            }
        });
    }
}
